use std::fmt::Debug;
use std::sync::Arc;
use async_trait::async_trait;
use anyhow::Result;
use serde::{Deserialize, Serialize};
use sqlx::{PgPool, Pool, Postgres, Row};
use sqlx::postgres::PgQueryResult;
use tracing::{debug, error, info, warn};

use crate::persistence::{EventStore, PersistenceError, PersistentEvent, PersistentState};

/// PostgreSQL-based event store implementation
pub struct PostgresEventStore<E, S>
where
    E: PersistentEvent + for<'de> Deserialize<'de> + Serialize + Clone + Debug + 'static,
    S: PersistentState + for<'de> Deserialize<'de> + Serialize + Clone + Debug + 'static,
{
    /// Database connection pool
    pool: Pool<Postgres>,
    
    /// Schema name for tables
    schema: String,
    
    /// Events table name
    events_table: String,
    
    /// Snapshots table name
    snapshots_table: String,
    
    /// Interval between snapshots (0 means no snapshots)
    snapshot_interval: u64,
    
    /// Phantom data for generics
    _marker_e: std::marker::PhantomData<E>,
    _marker_s: std::marker::PhantomData<S>,
}

impl<E, S> PostgresEventStore<E, S>
where
    E: PersistentEvent + for<'de> Deserialize<'de> + Serialize + Clone + Debug + 'static,
    S: PersistentState + for<'de> Deserialize<'de> + Serialize + Clone + Debug + 'static,
{
    /// Create a new PostgreSQL event store
    pub async fn new(
        connection_string: &str,
        schema: &str,
        events_table: &str,
        snapshots_table: &str,
        snapshot_interval: u64,
    ) -> Result<Self, PersistenceError> {
        // 连接数据库
        let pool = PgPool::connect(connection_string)
            .await
            .map_err(|e| PersistenceError::General(e.into()))?;
        
        let store = Self {
            pool,
            schema: schema.to_string(),
            events_table: events_table.to_string(),
            snapshots_table: snapshots_table.to_string(),
            snapshot_interval,
            _marker_e: std::marker::PhantomData,
            _marker_s: std::marker::PhantomData,
        };
        
        // 初始化数据库表
        store.init_tables().await?;
        
        Ok(store)
    }
    
    /// 初始化数据库表
    async fn init_tables(&self) -> Result<(), PersistenceError> {
        // 创建schema（如果不存在）
        let create_schema = format!("CREATE SCHEMA IF NOT EXISTS {}", self.schema);
        sqlx::query(&create_schema)
            .execute(&self.pool)
            .await
            .map_err(|e| PersistenceError::General(e.into()))?;
        
        // 创建事件表
        let create_events_table = format!(
            "CREATE TABLE IF NOT EXISTS {}.{} (
                actor_id VARCHAR(255) NOT NULL,
                index BIGINT NOT NULL,
                event_type VARCHAR(255) NOT NULL,
                data JSONB NOT NULL,
                created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP,
                PRIMARY KEY (actor_id, index)
            )",
            self.schema, self.events_table
        );
        sqlx::query(&create_events_table)
            .execute(&self.pool)
            .await
            .map_err(|e| PersistenceError::General(e.into()))?;
        
        // 创建快照表
        let create_snapshots_table = format!(
            "CREATE TABLE IF NOT EXISTS {}.{} (
                actor_id VARCHAR(255) NOT NULL,
                version BIGINT NOT NULL,
                data JSONB NOT NULL,
                created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP,
                PRIMARY KEY (actor_id)
            )",
            self.schema, self.snapshots_table
        );
        sqlx::query(&create_snapshots_table)
            .execute(&self.pool)
            .await
            .map_err(|e| PersistenceError::General(e.into()))?;
        
        Ok(())
    }
}

#[async_trait]
impl<E, S> EventStore for PostgresEventStore<E, S>
where
    E: PersistentEvent + for<'de> Deserialize<'de> + Serialize + Clone + Debug + 'static,
    S: PersistentState + for<'de> Deserialize<'de> + Serialize + Clone + Debug + 'static,
{
    type Event = E;
    type State = S;
    
    fn snapshot_interval(&self) -> u64 {
        self.snapshot_interval
    }
    
    async fn save_event(&self, actor_id: &str, event: &Self::Event, index: u64) -> Result<(), PersistenceError> {
        let event_type = event.event_type();
        let data = serde_json::to_value(event)
            .map_err(|e| PersistenceError::WriteError(format!("Failed to serialize event: {}", e)))?;
        
        let query = format!(
            "INSERT INTO {}.{} (actor_id, index, event_type, data) VALUES ($1, $2, $3, $4)",
            self.schema, self.events_table
        );
        
        sqlx::query(&query)
            .bind(actor_id)
            .bind(index as i64)
            .bind(event_type)
            .bind(data)
            .execute(&self.pool)
            .await
            .map_err(|e| PersistenceError::WriteError(format!("Failed to insert event: {}", e)))?;
        
        debug!("Saved event for actor {}, index {}, type {}", actor_id, index, event_type);
        Ok(())
    }
    
    async fn get_events(&self, actor_id: &str, start_index: u64) -> Result<Vec<Self::Event>, PersistenceError> {
        let query = format!(
            "SELECT data FROM {}.{} WHERE actor_id = $1 AND index >= $2 ORDER BY index ASC",
            self.schema, self.events_table
        );
        
        let rows = sqlx::query(&query)
            .bind(actor_id)
            .bind(start_index as i64)
            .fetch_all(&self.pool)
            .await
            .map_err(|e| PersistenceError::ReadError(format!("Failed to read events: {}", e)))?;
        
        let mut events = Vec::with_capacity(rows.len());
        for row in rows {
            let data: serde_json::Value = row.get("data");
            let event = serde_json::from_value(data)
                .map_err(|e| PersistenceError::ReadError(format!("Failed to deserialize event: {}", e)))?;
            events.push(event);
        }
        
        debug!("Retrieved {} events for actor {} from index {}", events.len(), actor_id, start_index);
        Ok(events)
    }
    
    async fn save_snapshot(&self, actor_id: &str, state: &Self::State) -> Result<(), PersistenceError> {
        let version = state.version();
        let data = serde_json::to_value(state)
            .map_err(|e| PersistenceError::SnapshotWriteError(format!("Failed to serialize state: {}", e)))?;
        
        let query = format!(
            "INSERT INTO {}.{} (actor_id, version, data) 
             VALUES ($1, $2, $3)
             ON CONFLICT (actor_id) 
             DO UPDATE SET version = $2, data = $3, created_at = CURRENT_TIMESTAMP",
            self.schema, self.snapshots_table
        );
        
        sqlx::query(&query)
            .bind(actor_id)
            .bind(version as i64)
            .bind(data)
            .execute(&self.pool)
            .await
            .map_err(|e| PersistenceError::SnapshotWriteError(format!("Failed to update snapshot: {}", e)))?;
        
        debug!("Saved snapshot for actor {}, version {}", actor_id, version);
        Ok(())
    }
    
    async fn get_snapshot(&self, actor_id: &str) -> Result<Option<Self::State>, PersistenceError> {
        let query = format!(
            "SELECT data FROM {}.{} WHERE actor_id = $1",
            self.schema, self.snapshots_table
        );
        
        let row = sqlx::query(&query)
            .bind(actor_id)
            .fetch_optional(&self.pool)
            .await
            .map_err(|e| PersistenceError::SnapshotReadError(format!("Failed to read snapshot: {}", e)))?;
        
        if let Some(row) = row {
            let data: serde_json::Value = row.get("data");
            let state = serde_json::from_value(data)
                .map_err(|e| PersistenceError::SnapshotReadError(format!("Failed to deserialize state: {}", e)))?;
            debug!("Retrieved snapshot for actor {}", actor_id);
            Ok(Some(state))
        } else {
            debug!("No snapshot found for actor {}", actor_id);
            Ok(None)
        }
    }
} 